package rabbitMQ.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitMQ.common.Common;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static final int MESSAGE_COUNT = 200;
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        //单独确认
//        publishingMessagesIndividually();
        //批量确认
//        publishingMessagesInBatches();
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException{
        try(Connection connection = createConnection()){
            Channel channel = connection.createChannel();
            channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);
            channel.confirmSelect();

            //有序集合, 元素按照自然顺序进行排序, 存储未confirm消息序号
            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple){
                        //批量
                        confirmSet.headSet(deliveryTag + 1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple){
                        //批量
                        confirmSet.headSet(deliveryTag + 1).clear();
                    }else{
                        confirmSet.remove(deliveryTag);
                    }
                    //如果处理失败, 需要有消息重发的环节, 此处省略
                }
            });
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = "消息" + i;
                long nextPublishSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE3, null, message.getBytes(StandardCharsets.UTF_8));
                confirmSet.add(nextPublishSeqNo);
            }
            while(!confirmSet.isEmpty()){
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.printf("Published %d message ConfirmsAsynchronously in %d ms", MESSAGE_COUNT, end - start);
        }
    }

    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            Channel channel = connection.createChannel();
            channel.confirmSelect();
            channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);
            //批量个数
            int batchSize = 100;
            int outstandingMessageCount = 0;
            long start = System.currentTimeMillis();
            for(int i = 0; i < MESSAGE_COUNT; i++){
                String body = "消息" + i;
                channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE2, null, body.getBytes(StandardCharsets.UTF_8));
                outstandingMessageCount++;
                if(outstandingMessageCount == batchSize){
                    channel.waitForConfirmsOrDie(5000);
                    outstandingMessageCount = 0;
                }
            }
            if(outstandingMessageCount > 0){
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("Published %d message batch in %d ms", MESSAGE_COUNT, end - start);
        }
    }

    public static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            //创建channel
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();
            channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE1, true, false, false, null);
            Long start = System.currentTimeMillis();
            for(int i = 0; i < MESSAGE_COUNT; i++){
                String body = "消息" + i;
                channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE1, null, body.getBytes(StandardCharsets.UTF_8));
                //等待确认消息, 只要消息被确认, 这个方法就会被返回
                //如果超时过期, 则抛出TimeoutException, 如果任何消息被nack(丢失), waitForConfirmsOrDie将抛出IOException
                channel.waitForConfirmsOrDie(5000);

            }
            Long end = System.currentTimeMillis();
            System.out.printf("Published %d message individually in %d ms", MESSAGE_COUNT, end - start);
        }

    }

    public static Connection createConnection() throws IOException, TimeoutException {
        //1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2. 设置参数
        factory.setHost(Common.HOST);//ip
        factory.setPort(Common.PORT);//端口号
        factory.setVirtualHost(Common.VIRTUALHOST);//虚拟机名称
        factory.setUsername(Common.USERNAME);//用户名
        factory.setPassword(Common.PASSWORD);//密码
        //3. 创建连接connection
        return factory.newConnection();
    }
}
